package com.ruoyi.web.websocket;

import com.alibaba.fastjson2.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.JsonObject;
import com.ruoyi.closesession.domain.CloseSession;
import com.ruoyi.closesession.mapper.CloseSessionMapper;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.conversation.domain.Conversation;
import com.ruoyi.conversation.mapper.ConversationMapper;
import com.ruoyi.distributionrule.domain.DistributionRule;
import com.ruoyi.distributionrule.mapper.DistributionRuleMapper;
import com.ruoyi.finance.domain.Finance;
import com.ruoyi.finance.mapper.FinanceMapper;
import com.ruoyi.order.domain.Order;
import com.ruoyi.order.mapper.OrderMapper;
import com.ruoyi.user.domain.User;
import com.ruoyi.user.domain.vo.UserVO;
import com.ruoyi.user.service.IUserService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.*;

/**
 * websocket 消息处理类（支持多端多会话 + 离线消息持久化）
 */
@Api(tags = "聊天会话WebSocket接口")
@Component
@ServerEndpoint("/wx/message")
public class WebSocketServer {

    private final ConversationMapper conversationMapper = SpringContextHolder.getBean(ConversationMapper.class);
    private final OrderMapper orderMapper = SpringContextHolder.getBean(OrderMapper.class);
    private final ObjectMapper objectMapper = SpringContextHolder.getBean(ObjectMapper.class);
    private final CloseSessionMapper closeSessionMapper = SpringContextHolder.getBean(CloseSessionMapper.class);
    private final IUserService userService = SpringContextHolder.getBean(IUserService.class);
    private final FinanceMapper financeMapper = SpringContextHolder.getBean(FinanceMapper.class);
    private final DistributionRuleMapper distributionRuleMapper = SpringContextHolder.getBean(DistributionRuleMapper.class);

    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);

    public static int socketMaxOnlineCount = 100;
    private static final Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);
    private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private static final Map<Session, Long> lastPingMap = new ConcurrentHashMap<>();
    private static final Map<String, ScheduledFuture<?>> pendingCloseTasks = new ConcurrentHashMap<>();

    /** key: orderId_userId，value: sessionId */
    private static final Map<String, String> sessionMap = new ConcurrentHashMap<>();

    /** key: sessionId，value: Session */
    private static final Map<String, Session> idSessionMap = new ConcurrentHashMap<>();

    // 定期清理超时未心跳的连接（每30秒执行一次）
    static {
        scheduler.scheduleAtFixedRate(() -> {
            long now = System.currentTimeMillis();
            for (Map.Entry<Session, Long> entry : lastPingMap.entrySet()) {
                Session session = entry.getKey();
                long lastPingTime = entry.getValue();
                if (now - lastPingTime > 90_000) { // 超过90秒无心跳
                    String openId = WebSocketUsers.getOpenIdBySession(session);
                    LOGGER.warn("会话超时未发送心跳，主动断开 - {} [{}]", openId, session.getId());
                    closeSafely(session);
                }
            }
        }, 30, 30, TimeUnit.SECONDS);
    }

    @ApiOperation("建立 WebSocket 连接")
    @OnOpen
    public void onOpen(Session session) throws IOException {
        if (!SemaphoreUtils.tryAcquire(socketSemaphore)) {
            LOGGER.error("当前在线人数超过限制: {}", socketMaxOnlineCount);
            sendMessage(session, MessageResult.error("当前在线人数超过限制：" + socketMaxOnlineCount));
            session.close();
            return;
        }

        Map<String, List<String>> params = session.getRequestParameterMap();
        List<String> openIds = params.get("openId");
        List<String> orderIds = params.get("orderId");

        if (openIds != null && !openIds.isEmpty() && orderIds != null && !orderIds.isEmpty()) {
            String openId = openIds.get(0);
            String orderId = orderIds.get(0);

            // 检查订单是否已结束
            Order order = orderMapper.selectOrderByOrderId(orderId);
            if (order != null && order.getIsSession() != null && order.getIsSession() == 1) {
                LOGGER.warn("订单 {} 会话已结束，拒绝用户 {} 建立连接", orderId, openId);
                sendMessage(session, MessageResult.error("订单会话已结束，无法建立连接"));
                session.close();
                SemaphoreUtils.release(socketSemaphore);
                return;
            }

            String key = orderId + "_" + openId;

            // 判断是否已有旧连接，存在则关闭
            if (sessionMap.containsKey(key)) {
                String oldSessionId = sessionMap.get(key);
                Session oldSession = idSessionMap.get(oldSessionId);
                if (oldSession != null && oldSession.isOpen()) {
                    LOGGER.info("检测到重复连接，关闭旧连接 sessionId={}", oldSessionId);
                    closeSafely(oldSession);
                }
                // 清理旧连接映射
                sessionMap.remove(key);
                idSessionMap.remove(oldSessionId);
            }

            // 添加新连接
            WebSocketUsers.put(openId, session);
            lastPingMap.put(session, System.currentTimeMillis());

            sessionMap.put(key, session.getId());
            idSessionMap.put(session.getId(), session);

            LOGGER.info("用户上线 - {} [{}], key={}", openId, session.getId(), key);

            // 查询离线消息逻辑（注释部分可自行开启）
        /*
        Conversation conversation = new Conversation();
        conversation.setToUser(openId);
        conversation.setIsOffline(0);
        List<Conversation> offlineMessages = conversationMapper.selectConversationList(conversation);
        if (offlineMessages != null && !offlineMessages.isEmpty()) {
            for (Conversation offlineMsg : offlineMessages) {
                sendMessage(session, offlineMsg);
                offlineMsg.setIsOffline(1);
                conversationMapper.updateConversation(offlineMsg);
            }
            LOGGER.info("推送并更新用户 {} 的离线消息数量: {}", openId, offlineMessages.size());
        }
        */

            LOGGER.info("当前该用户会话数 - {}", WebSocketUsers.get(openId).size());
            LOGGER.info("当前在线人数（所有会话） - {}", WebSocketUsers.getUsers().values().stream().mapToInt(Set::size).sum());
            sendMessage(session, MessageResult.success("连接成功"));
        } else {
            LOGGER.warn("连接缺少 openId 或 orderId 参数");
            sendMessage(session, MessageResult.error("缺少 openId 或 orderId 参数"));
            session.close();
            SemaphoreUtils.release(socketSemaphore);
        }
    }


    @ApiOperation("关闭 WebSocket 连接")
    @OnClose
    public void onClose(Session session) {
        LOGGER.info("关闭连接 - [{}]", session.getId());
        String openId = WebSocketUsers.removeBySession(session);
        if (openId != null) {
            LOGGER.info("当前该用户剩余会话数 - {}", WebSocketUsers.get(openId).size());
        }
        lastPingMap.remove(session);
        SemaphoreUtils.release(socketSemaphore);

        // 移除 sessionMap 和 idSessionMap 中对应的数据
        removeSessionFromMaps(session);
    }

    @ApiOperation("WebSocket 连接出现异常")
    @OnError
    public void onError(Session session, Throwable exception) {
        LOGGER.error("WebSocket 连接异常 [{}]: {}", session.getId(), exception.getMessage(), exception);
        closeSafely(session);
    }

    @ApiOperation("接收客户端消息")
    @OnMessage
    public void onMessage(String msg, Session session) {
        // 心跳检测处理
        if ("ping".equalsIgnoreCase(msg.trim())) {
            LOGGER.debug("收到心跳 ping [{}]", session.getId());
            lastPingMap.put(session, System.currentTimeMillis());
            sendMessage(session, "pong");
            return;
        }

        try {
            Conversation message = objectMapper.readValue(msg, Conversation.class);
            LOGGER.info("收到消息 - {}", message);

            if (message.getOrderId() == null || message.getOrderId().trim().isEmpty()) {
                LOGGER.warn("消息缺少 orderId，已丢弃: {}", message);
                sendMessage(session, MessageResult.error("缺少 orderId"));
                return;
            }

            if (message.getFromUser() == null || message.getFromUser().trim().isEmpty()) {
                List<String> openIds = session.getRequestParameterMap().get("openId");
                if (openIds != null && !openIds.isEmpty()) {
                    message.setFromUser(openIds.get(0));
                }
            }

            message.setCreateTime(new Date());
            message.setSessionId(session.getId());
            conversationMapper.insertConversation(message);
            LOGGER.info("消息已入库 - {}", message);

            switch (message.getActionType()) {
                case "message":
                    // 用户有发送新消息，则取消之前的自动关闭任务
                    ScheduledFuture<?> task = pendingCloseTasks.remove(message.getOrderId());
                    if (task != null) {
                        task.cancel(false);
                        LOGGER.info("用户 {} 继续发消息，取消订单 {} 的自动关闭任务", message.getFromUser(), message.getOrderId());
                    }
                    forwardMessage(message);
                    break;
                case "closeRequest":
                    handleCloseRequest(message);
                    break;
                case "closeConfirm":
                    handleCloseConfirm(message.getOrderId());
                    break;
                default:
                    LOGGER.warn("未知 actionType: {}", message.getActionType());
                    sendMessage(session, MessageResult.error("未知 actionType: " + message.getActionType()));
            }
        } catch (IOException e) {
            LOGGER.error("消息解析失败: {}", msg, e);
            sendMessage(session, MessageResult.error("消息格式错误"));
        }
    }

    /**
     * 消息转发（区分在线与离线）
     */
    private void forwardMessage(Conversation message) {
        Set<Session> sessions = WebSocketUsers.get(message.getToUser());
        if (sessions != null && !sessions.isEmpty()) {
            message.setIsOffline(1); // 在线消息
            for (Session s : sessions) {
                sendMessage(s, message);
            }
            conversationMapper.updateConversation(message);
            LOGGER.info("消息已发送给在线用户 {}", message.getToUser());
        } else {
            message.setIsOffline(0); // 离线消息
            conversationMapper.updateConversation(message);
            LOGGER.warn("目标用户 {} 不在线，消息已保存为离线消息", message.getToUser());
        }
    }

    private void handleCloseRequest(Conversation message) {
        if ("user".equals(message.getFromRole())) {
            closeOrder(message.getOrderId(), "用户已结束会话");
        } else if ("expert".equals(message.getFromRole())) {
            forwardMessage(message);
            CloseSession closeSession = closeSessionMapper.selectCloseSessionById(1L);
            String endSession = closeSession.getEndSession();
            long l = Long.parseLong(endSession);
            LOGGER.info("专家请求结束，等待用户确认...");
            ScheduledFuture<?> task = scheduler.schedule(() ->
                    closeOrder(message.getOrderId(), "用户未响应，系统自动结束会话"), l, TimeUnit.HOURS);
            pendingCloseTasks.put(message.getOrderId(), task);
        }
    }

    private void handleCloseConfirm(String orderId) {
        ScheduledFuture<?> task = pendingCloseTasks.remove(orderId);
        if (task != null) {
            task.cancel(false);
        }
        closeOrder(orderId, "用户已确认结束会话");
    }

    private void closeOrder(String orderId, String reason) {
        LOGGER.info("结束订单 {} - {}", orderId, reason);
        Order order = orderMapper.selectOrderByOrderId(orderId);
        if (order != null) {
            order.setIsSession(1);
            order.setOrderStatus("0");
            order.setIsEvaluate("1");
            order.setCompleteTime(DateUtils.getNowDate());
            orderMapper.updateOrder(order);
            UserVO userVO = userService.selectUserByUserId(order.getExpertId());
            User user = new User();
            BeanUtils.copyProperties(userVO, user);
            order = orderMapper.selectOrderByOrderId(order.getOrderId());
            BigDecimal freezeAmount = user.getFreezeAmount();
            DistributionRule distributionRule = new DistributionRule();
            distributionRule = distributionRuleMapper.selectDistributionRuleList(distributionRule).get(0);
            String orderCommission = distributionRule.getOrderCommission();
            BigDecimal commissionRate = new BigDecimal(orderCommission).divide(BigDecimal.valueOf(100));
            BigDecimal actualPayment = order.getActualPayment();
            BigDecimal orderMultiply = actualPayment.multiply(commissionRate);
            freezeAmount = freezeAmount.subtract(orderMultiply);
            user.setWithdrawableAmount(user.getWithdrawableAmount().add(orderMultiply));
            user.setFreezeAmount(freezeAmount);
            userService.updateUser(user);
            String expertCommission = distributionRule.getExpertCommission();
            BigDecimal expertCommissionRate = new BigDecimal(expertCommission).divide(BigDecimal.valueOf(100));
            String inviterId = user.getInviterId();
            UserVO userVO1 = userService.selectUserByUserId(inviterId);
            if (userVO1 != null) {
                User user1 = new User();
                BeanUtils.copyProperties(userVO1, user1);
                BigDecimal freezeAmount1 = user1.getFreezeAmount();
                BigDecimal actualPayment1 = order.getActualPayment();
                BigDecimal multiply = actualPayment1.multiply(expertCommissionRate);
                BigDecimal subtract = freezeAmount1.subtract(multiply);
                user1.setWithdrawableAmount(user1.getWithdrawableAmount().add(multiply));
                user1.setFreezeAmount(subtract);
                userService.updateUser(user1);
            }
            String userCommission = distributionRule.getUserCommission();
            BigDecimal userCommissionRate = new BigDecimal(userCommission).divide(BigDecimal.valueOf(100));
            UserVO userVO2 = userService.selectUserByUserId(order.getUserId());
            String inviterId1 = userVO2.getInviterId();
            UserVO userVO3 = userService.selectUserByUserId(inviterId1);
            if (userVO3 != null) {
                User user3 = new User();
                BeanUtils.copyProperties(userVO3, user3);
                if (userVO3.getIsExpert() == 0) {
                    BigDecimal freezeAmount1 = user3.getFreezeAmount();
                    BigDecimal actualPayment1 = order.getActualPayment();
                    BigDecimal multiply = actualPayment1.multiply(userCommissionRate);
                    BigDecimal subtract = freezeAmount1.subtract(multiply);
                    user3.setWithdrawableAmount(user3.getWithdrawableAmount().add(multiply));
                    user3.setFreezeAmount(subtract);
                    userService.updateUser(user3);
                }
            }
            Finance finance = new Finance();
            finance.setOrderId(orderId);
            List<Finance> finances = financeMapper.selectFinanceList(finance);
            if (!finances.isEmpty()) {
                Finance finance1 = finances.get(0);
                finance1.setSettlementTime(DateUtils.getNowDate());
                financeMapper.updateFinance(finance1);
            }
        }

        Conversation endMsg = new Conversation();
        endMsg.setActionType("closeNotice");
        endMsg.setContent(reason);
        endMsg.setOrderId(orderId);
        endMsg.setCreateTime(new Date());

        WebSocketUsers.getUsers().values().forEach(sessions ->
                sessions.forEach(session -> sendMessage(session, endMsg)));
    }

    private void sendMessage(Session session, Object message) {
        try {
            if (session != null && session.isOpen()) {
                String json = objectMapper.writeValueAsString(message);
                session.getBasicRemote().sendText(json);
            }
        } catch (IOException e) {
            LOGGER.warn("发送消息失败 [{}]: {}, 连接可能已关闭", session.getId(), message);
            // 连接异常时主动关闭清理
            closeSafely(session);
        }
    }


    private static void closeSafely(Session session) {
        try {
            if (session.isOpen()) {
                session.close();
            }
            String openId = WebSocketUsers.removeBySession(session);
            lastPingMap.remove(session);
            SemaphoreUtils.release(socketSemaphore);

            removeSessionFromMaps(session);

            LOGGER.info("安全关闭 session [{}]，openId: {}", session.getId(), openId);
        } catch (IOException e) {
            LOGGER.error("关闭 session 异常 [{}]", session.getId(), e);
        }
    }

    private static void removeSessionFromMaps(Session session) {
        idSessionMap.remove(session.getId());
        sessionMap.entrySet().removeIf(entry -> entry.getValue().equals(session.getId()));
    }

    /**
     * 关闭指定订单号和用户ID对应的 WebSocket 会话
     */
    public void closeSessionByOrderAndUser(String orderId, String userId, String reason) {
        String key = orderId + "_" + userId;
        String sessionId = sessionMap.get(key);
        if (sessionId != null) {
            Session session = idSessionMap.get(sessionId);
            if (session != null && session.isOpen()) {
                sendMessage(session, MessageResult.error("连接被关闭：" + reason));
                closeSafely(session);
                LOGGER.info("已关闭订单 {} 用户 {} 的会话", orderId, userId);
            } else {
                LOGGER.warn("Session 已关闭或不存在，移除缓存 key: {}", key);
            }
            sessionMap.remove(key);
            idSessionMap.remove(sessionId);
        } else {
            LOGGER.warn("找不到订单 {} 用户 {} 的会话", orderId, userId);
        }
    }

    /**
     * 关闭两个用户所有 WebSocket 连接
     */
    public void closeBothUsers(String userA, String userB, String reason) {
        LOGGER.info("准备关闭用户 {} 和 {} 的所有连接，原因: {}", userA, userB, reason);
        closeUserSessions(userA, reason);
        closeUserSessions(userB, reason);
    }

    /**
     * 关闭指定用户所有 WebSocket 会话
     */
    private void closeUserSessions(String openId, String reason) {
        Set<Session> sessions = WebSocketUsers.get(openId);
        if (sessions != null && !sessions.isEmpty()) {
            for (Session session : sessions) {
                sendMessage(session, MessageResult.error("连接被关闭：" + reason));
                closeSafely(session);
            }
            LOGGER.info("已关闭用户 {} 的所有 {} 个会话", openId, sessions.size());
        } else {
            LOGGER.warn("用户 {} 无在线会话可关闭", openId);
        }
    }
}
